package cn.edu.flink.scala.tutorial.sql

import cn.edu.flink.scala.tutorial.source.ClickSource
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Expressions.$
import org.apache.flink.table.api.bridge.scala._

object FileTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)

    val clickStream = env.addSource(new ClickSource).assignAscendingTimestamps(_.timestamp)

    val tEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)
    val click_table = tEnv.fromDataStream(clickStream,
      $("user"),
      $("url"),
      $("ts").rowtime())
    tEnv.createTemporaryView("click_table", click_table)

    tEnv.executeSql(
      s"""
         |create table output_click_table (
         |user_name string,
         |url string,
         |ts timestamp
         |)with(
         |'connector' = 'print'
         |--'path' = 'src/main/resources/click_table.txt',
         |--'format' = 'json'
         |)
         |""".stripMargin)

    tEnv.executeSql(
      s"""
         |insert into output_click_table
         |select user, url, ts
         |from click_table
         |""".stripMargin)

    env.execute("FileTest")
  }
}
